package csv

import (
	"encoding/csv"
	"fmt"
	"io"
	"os"

	"github.com/example-user/go-pocket-etl/pkg/core"
	"github.com/example-user/go-pocket-etl/pkg/factory"
)

// init 实现了组件的自注册，在程序启动时自动将 CSV Source 添加到工厂。
func init() {
	factory.RegisterSource("csv", NewSource)
}

// Source 实现了 core.Source 接口，用于从 CSV 文件读取数据。
type Source struct {
	filePath  string      // 要读取的CSV文件路径
	file      *os.File    // 文件句柄
	reader    *csv.Reader // Go 标准库的 CSV 读取器
	hasHeader bool        // 标记CSV是否包含表头
	header    []string    // 如果有表头，存储表头列名
	delimiter rune        // CSV文件的分隔符，默认为逗号
	line      int         // 当前已读取的行数，用于精确的错误报告
}

// NewSource 是 csv.Source 的构造函数，由工厂调用。
func NewSource() core.Source {
	return &Source{
		delimiter: ',', // 默认为逗号
	}
}

// Open 负责解析配置、打开 CSV 文件并准备读取。
// 它会处理文件路径、是否包含表头、自定义分隔符等配置项。
// 如果配置了 has_header: true，它会预先读取第一行作为后续 Record 的键。
func (s *Source) Open(config map[string]interface{}) error {
	// 'file_path' 是必需配置。
	filePath, ok := config["file_path"].(string)
	if !ok {
		return fmt.Errorf("csv source: config is missing required key 'file_path'")
	}
	s.filePath = filePath

	// 'has_header' 是可选配置，默认为 true。
	if hasHeader, ok := config["has_header"].(bool); ok {
		s.hasHeader = hasHeader
	} else {
		s.hasHeader = true
	}

	// 'delimiter' 是可选配置，默认为 ','。
	if delimiterStr, ok := config["delimiter"].(string); ok && len(delimiterStr) > 0 {
		// 只取第一个字符作为分隔符。
		s.delimiter = []rune(delimiterStr)[0]
	}

	var err error
	s.file, err = os.Open(s.filePath)
	if err != nil {
		return fmt.Errorf("csv source: failed to open file %s: %w", s.filePath, err)
	}

	s.reader = csv.NewReader(s.file)
	s.reader.Comma = s.delimiter

	s.line = 0

	// 如果有表头，立即读取并存储，以便后续的 Read() 调用使用。
	if s.hasHeader {
		s.header, err = s.reader.Read()
		if err != nil {
			if err == io.EOF {
				return fmt.Errorf("csv source: file is empty or contains only a header")
			}
			return fmt.Errorf("csv source: failed to read header: %w", err)
		}
		s.line++
	}

	return nil
}

// Read 读取 CSV 文件中的下一行，并将其转换为一个 core.Record。
// 如果文件定义了表头，则使用表头作为键；否则，自动生成 "column_1", "column_2", ... 作为键。
// 它还会校验每行数据的列数是否与表头匹配，以确保数据规整。
func (s *Source) Read() (core.Record, error) {
	row, err := s.reader.Read()
	if err != nil {
		if err == io.EOF {
			return nil, io.EOF // 这是数据流正常结束的信号
		}
		return nil, fmt.Errorf("csv source: error reading data at line %d: %w", s.line+1, err)
	}
	s.line++

	record := make(core.Record)

	if s.hasHeader {
		// 关键的数据完整性校验：确保每行数据的列数与表头一致。
		if len(row) != len(s.header) {
			return nil, fmt.Errorf("csv source: column count mismatch at line %d. Expected %d, got %d", s.line, len(s.header), len(row))
		}
		for i, value := range row {
			record[s.header[i]] = value
		}
	} else {
		// 当没有表头时，自动生成列名。
		for i, value := range row {
			key := fmt.Sprintf("column_%d", i+1)
			record[key] = value
		}
	}

	return record, nil
}

// Close 实现了 core.Source 接口，负责关闭已打开的文件句柄，释放资源。
func (s *Source) Close() error {
	if s.file != nil {
		return s.file.Close()
	}
	return nil
}
